package com.doit.day04

import scala.io.Source

object _07_电影案例 {
  def main(args: Array[String]): Unit = {

    //1.读取数据
    val movies: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\movies.txt").getLines().toList
    val ratings: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\ratings.txt").getLines().toList
    val users: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\users.txt").getLines().toList

    //2.转换格式，搞成map集合或者list里面放的是元祖
    //(电影id , (电影名称  , 电影类型))
    val movies_map: Map[String, (String, String)] = movies.map(line => {
      //1_Toy Story (1995)_Animation|Children's|Comedy
      val arr: Array[String] = line.split("_")
      //电影id  电影名称   电影类型
      (arr(0), (arr(1), arr(2)))
    }).toMap

    //(用户id，电影id，评分，时间戳)
    val ratings_list: List[(String, String, String, String)] = ratings.map(line => {
      //6039_2208_4_956706182
      val arr: Array[String] = line.split("_")
      (arr(0), arr(1), arr(2), arr(3))
    })

    //用户id，性别，年龄，职业，邮编
    val users_list: Map[String, (String, String, String, String)] = users.map(line => {
      //1_F_1_10_48067
      val arr: Array[String] = line.split("_")
      (arr(0), (arr(1), arr(2), arr(3), arr(4)))
    }).toMap

   /* xuQui2(movies_map,ratings_list)
      .toList
      .sortBy(_._2)
      .reverse
      .take(10)
      .foreach(println)*/

 /*   xuQui4(ratings_list,users_list)
      .toList
      .sortBy(_._1)
      .foreach(println)
*/

    xuQui5(ratings_list,users_list)
      .foreach(println)

  }

  //1.求每部电影的平均分，按照降序输出 取前20个  ==> 我需要得是电影名称 + 平均分
  def xuQui1(movies_map: Map[String, (String, String)],ratings_list: List[(String, String, String, String)])={
    //对评分表中的电影id进行分组
    //key 电影id    value 是所有和电影id相同的元祖都放在一个集合中   接下来你要干啥？？？
    val movie_idGroupEd: Map[String, List[(String, String, String, String)]] = ratings_list.groupBy(_._2)
    movie_idGroupEd.map(tp=>{
      val movie_id: String = tp._1
      val ratings: List[Double] = tp._2.map(_._3.toDouble)
      val avg: Double = (ratings.sum / tp._2.size).formatted("%.2f").toDouble
      //根据电影id获取对应的电影名称
      val movie_name: String = movies_map.getOrElse(movie_id, ("找不到", "找不到"))._1
      (movie_name,avg)
    })
  }

  //2.每个电影类型的平均分  ==》 按降序排列 取前10个
  //看着好像和第一个需求挺像，但是一个电影对应多个类型
  def xuQui2(movies_map: Map[String, (String, String)],ratings_list: List[(String, String, String, String)])={

    val idAndTypes: Map[String, Array[String]] = movies_map.map(tp => {
      val movie_id: String = tp._1
      val arr_movie_types: Array[String] = tp._2._2.split("\\|")
      (movie_id, arr_movie_types)
    })

    val flatten: List[(String, Double)] = ratings_list.map(tp => {
      val movie_id: String = tp._2
      val movie_types: Array[String] = idAndTypes.getOrElse(movie_id, Array(""))
      //将每个电影类型都获取出来，和分数进行拼接
      movie_types.map((_, tp._3.toDouble))
    }).flatten

    flatten.groupBy(_._1)
      .map(tp=>(tp._1,(tp._2.map(_._2).sum/tp._2.size).formatted("%.2f").toDouble))

    //现在有的是什么  (电影id,分数)==> 通过两个表的关联 ==》 电影类型，分数
    /* Animation|Children's|Comedy 3  ==》切割 Array(Animation，Children's，Comedy)  3
    Array(Animation，Children's，Comedy)  3==》 Array((Animation,3)，(Children's,3)，(Comedy,3))
    变成
        Animation 3
        Children's 3
        Comedy 3
     */
    //每个电影类型的平均分  如果说我能得到  List(电影类型，分数)

  }

  //3.哪些年份的电影评分(平均分)最高，取最高的前五个年份
  //1_Toy Story (1995)_Animation|Children's|Comedy
  def xuQui3(movies_map: Map[String, (String, String)],ratings_list: List[(String, String, String, String)])={

    //现在有什么？电影id 电影名称(年份) ==》 关联  电影id 分数

    //想要得到结果就需要list(年份，分数)

  }

  //4.每个职业最喜欢的前三个电影id   观看次数最多的
  def xuQui4(ratings_list: List[(String, String, String, String)],users_list: Map[String, (String, String, String, String)])={
    //根据电影评分表 ==》 可以对用户进行分组  得到每一个用户看过哪些电影
   /* val userGroupEd: List[(String, List[(String, String, String, String)])] = ratings_list.groupBy(_._1).toList
    val jobAndMovieId: List[(String, List[(String, String, String, String)])] = userGroupEd.map(tp => {
      val user_id: String = tp._1

      val job: String = users_list.getOrElse(user_id, ("未知", "未知", "未知", "未知"))._3

      (job, tp._2)
    })
    //拿这个结果对于职业进行关联，得到每一个用户的职业，职业看过哪些电影 ==》 Map 职业  List((职业，list[]))
    val jobAndMovies: Map[String, List[(String, List[(String, String, String, String)])]] = jobAndMovieId.groupBy(_._1)
    jobAndMovies.map(tp=>{
      val job: String = tp._1
      val movies: List[String] = tp._2.map(_._2.map(_._2)).flatten
      val tuples: List[(String, Int)] = movies.groupBy(word => word).map(tp => (tp._1, tp._2.size))
        .toList
        .sortBy(_._2)
        .reverse
        .take(3)
      (job,tuples)
    })*/

    //要职业和电影id==》如果我直接将两张表进行关联，把user_id换成职业
    val jobAndMovieId: List[(String, String)] = ratings_list.map(tp => {
      val user_id: String = tp._1
      val movie_id: String = tp._2
      val job: String = users_list.getOrElse(user_id, ("未知", "未知", "未知", "未知"))._3
      (job, movie_id)
    })

    val jobGroupEd: Map[String, List[(String, String)]] = jobAndMovieId.groupBy(_._1)


    jobGroupEd.map(tp=>{
      val job: String = tp._1
      val value: List[String]  = tp._2.map(_._2)
      val movie_id_and_count: List[(String, Int)] = value.groupBy(word => word)
        .map(tp => (tp._1, tp._2.size))
        .toList
        .sortBy(_._2)
        .reverse
        .take(3)
      (job,movie_id_and_count)
    })
  }

  //5.年龄段在“18-34”的男性年轻人，最喜欢看哪10部电影（Id）==>喜欢看哪10部电影
  def xuQui5(ratings_list: List[(String, String, String, String)],users_list: Map[String, (String, String, String, String)])={
    //先让他们关联 得到每个用户年龄和性别  ==》 整体过滤
    //1,M,18,199
    //1,M,18,299
    //2,F,19,399
    //......
    val tuples: List[(String, String, String, String)] = ratings_list.map(tp => {
      val user_id: String = tp._1
      val movie_id: String = tp._2
      val tuple: (String, String, String, String) = users_list.getOrElse(user_id, ("未知", "未知", "未知", "未知"))
      (user_id, tuple._1, tuple._2, movie_id)
    })

    //1,M,18,199
    //1,M,18,299
    //1,M,25,299
    //2,M,18,199
    //过滤出来一些不符合条件的数据
    val filtered: List[(String, String, String, String)] = tuples.filter(tp => {
      tp._2.equals("M") && (tp._3.equals("18") || tp._3.equals("25"))
    })

    val movie_ids: List[String] = filtered.map(_._4)
    movie_ids.groupBy(word=>word)
      .map(tp=>(tp._1,tp._2.size))
      .toList
      .sortBy(_._2)
      .reverse
      .take(10)

  }



}
